package cn.me.alphamq.broker.event.spi.listener.broker;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.event.model.broker.PushMsgEvent;
import cn.me.alphamq.broker.store.SlaveCtxManager;
import cn.me.alphamq.broker.util.QueueIdUtil;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.entity.req.PushMsgReq;
import cn.me.alphamq.common.entity.resp.PushMsgResp;
import cn.me.alphamq.common.enumeration.*;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.remote.SyncFuture;
import cn.me.alphamq.common.store.SyncFutureManager;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * 推送消息的事件的监听器
 *
 * @author f
 */
public class PushMsgListener implements Listener<PushMsgEvent> {

    private static final Logger log = LoggerFactory.getLogger(PushMsgListener.class);

    @Override
    public void onReceive(PushMsgEvent event) throws Exception {
        PushMsgReq pushMsgReq = event.getPushMessageReq();
        ChannelHandlerContext ctx = event.getCtx();
        String msgId = pushMsgReq.getMsgId();

        // 将消息写入 commitLog 文件
        pushMsgReq.setQueueId(QueueIdUtil.getQueueId(pushMsgReq.getTopic(), pushMsgReq.getQueueId()));
        Container.getCommitLogAppendHandler().appendMsg(new Message(pushMsgReq));

        boolean isAsyncSend = msgId == null;
        boolean isMasterNode = BrokerRoleEnum.MASTER == Container.getClusterProperties().getBrokerRole();
        if (isMasterNode) {
            // 如果是主节点，则发送同步请求给从节点
            SlaveCtxManager.writeAndFlush(EventCodeEnum.PUSH_MSG_REQ, pushMsgReq);

            // 如果是异步发送，则无需等待从节点同步完毕
            if (isAsyncSend) {
                return;
            }

            // 如果没有从节点，则直接返回
            if (SlaveCtxManager.isEmpty()) {
                ctx.writeAndFlush(new TcpMsg(EventCodeEnum.PUSH_MSG_RESP, JSON.toJSONBytes(new PushMsgResp(msgId))));
                return;
            }

            SyncFuture syncFuture = new SyncFuture(msgId);
            SyncFutureManager.put(msgId, syncFuture);

            PushMsgResp pushMsgResp = new PushMsgResp(msgId);
            try {
                // 等待从节点同步，最长 3s TODO 之后可以通过配置文件来设置最长等待时间
                syncFuture.get(3, TimeUnit.SECONDS);
            } catch (Exception e) {
                pushMsgResp.setPushStatus(PushStatusEnum.FAIL.getCode());
                pushMsgResp.setDesc("Slave node sync fail! Sync task had Exception!");
                log.error("slave sync error is:", e);
            }

            // 响应返回给到客户端，完成主从复制链路效果
            ctx.writeAndFlush(new TcpMsg(EventCodeEnum.PUSH_MSG_RESP, JSON.toJSONBytes(pushMsgResp)));
        } else {
            // 如果是异步发送，则 从节点/单节点 无需返回响应
            if (isAsyncSend) {
                return;
            }

            // 从节点 给 主节点 返回响应/单节点 给 客户端 返回响应
            ctx.writeAndFlush(new TcpMsg(EventCodeEnum.PUSH_MSG_RESP, JSON.toJSONBytes(new PushMsgResp(msgId))));
        }
    }
}
